#ifndef CONNPOOL_H
#define CONNPOOL_H

#include <pthread.h>
#include "common.h"
#include "hbase_client.h"
#include "hdfs_client.h"
//#include "memlink_client.h"

#define CONN_POOL_HBASE_INDEX	0
#define CONN_POOL_HDFS_INDEX	1 
#define CONN_POOL_MEMLINK_INDEX	2

#define CONN_POOL_TYPE_NUM	2
#define CONN_POOL_TYPE_HBASE	CONN_POOL_HBASE_INDEX	
#define CONN_POOL_TYPE_HDFS	CONN_POOL_HDFS_INDEX
#define CONN_POOL_TYPE_MEMLINK	CONN_POOL_MEMLINK_INDEX

#define HBASE_CREATE	1
#define HBASE_GET		2
#define HBASE_PUT		3
#define HBASE_DEL		4

extern void signal_mask(int sig);
extern void signal_unmask(int sig);

struct hbase_conn
{
	HbaseOp *conn;
	struct ConnPoolItem *pool;
	hbase_conn(const char *ip, unsigned short port);
	int open();
	//int op(char *buf, void **ret);
	int close();
	int reactive();
    
    char hostip[16];
    unsigned short port;
};

struct hdfs_conn
{
    HdfsOp *conn;
	struct ConnPoolItem *pool;
    hdfs_conn(const char *ip, unsigned short port);
	int reactive();
	int open();
	//int op(char *cmd, void *buf, void **ret);
	int close();
    
    char hostip[16];
    unsigned short port;
};

/*
struct memlink_conn
{
	MemLink *conn;
	struct ConnPoolItem *pool;
	memlink_conn(const char *host, unsigned short rport, unsigned short wport);
	int open();
	//int op(char *cmd, void *buf, void **ret);
	int close();
	int reactive();
    
    char hostip[16];
    unsigned short rport;
    unsigned short wport;
};
*/
struct ConnPoolItem
{
	void *item;
	int index;
	int id;
	volatile char active;
	char idle;
	char type;
	struct ConnPoolItem *next;
	struct ConnPoolItemHead *head;
};

struct ConnPoolItemHead
{
	pthread_mutex_t m;
	char type;
	struct ConnPoolItem *root;
	int check();
};

struct reactive_queue_item
{
	struct ConnPoolItem *conn;
	struct reactive_queue_item *next;
};

struct reactive_queue_head
{
	pthread_mutex_t m;
	int num;
	struct reactive_queue_item *root;
	struct reactive_queue_item *tail;
};

struct notify_fds
{
	int reactive_fd[2];
	int memlink_timer_fd[2];
};

class ConnPool
{
public:
	ConnPool();
	~ConnPool();
	int init(const Config* config);
	void *get(int type);
	int put(int type, void *conn);
public:
	struct ConnPoolItemHead pool[CONN_POOL_TYPE_NUM];
	struct notify_fds *fds;
private:
    const Config* conf;
	pthread_t tid;
	int start_connpool_monitor();
	int set_memlink_timer();
};

#define set_inactive(h) (h)->pool->active = 0

ConnPool *init_connpool(const Config* conf);
void close_connpool(ConnPool *pool);

#endif
